package day1;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class WordCound01 {
    public static void main(String[] args) throws Exception{
        // 1、创建一个env对象
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  // 去掉前缀

        //env.setRuntimeMode(RuntimeExecutionMode.STREAMING);       //指定计算模式为流
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);           //指定计算模式为批
        //env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);         //自动
        //不设置的话默认是流模式e(RuntimeExecutionMode.STREAMING)

        // 2、创建数据
        DataStreamSource<String> dataStream = env.fromElements("hello,heihei,hello", "word,hello,word", "word,word,hello", "heihei,word,haha", "haha,word,word", "haha,heihei,word");
        
        // 3、转换处理
        SingleOutputStreamOperator<String> dataStream2 = dataStream.flatMap(new FlatMapFunction<String, String>() {

               @Override
               public void flatMap(String s, Collector<String> collector) throws Exception {
                   String[] arr = s.split(",");
                   for (String s1 : arr) {
                       collector.collect(s1);
                   }
               }
           }
        );

        DataStream<Tuple2<String, Integer>> dataStream3 = dataStream2.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value, 1);
            }
        });

        // 分组 聚合
        KeyedStream<Tuple2<String, Integer>, String> dataStream4 = dataStream3.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                // 根据 key 进行分组
                return value.f0;
            }
        });

        // 聚合(相加)操作
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = dataStream4.sum(1);
        // result.print(); // 输出到 8081 Task Managers
        result.writeAsCsv("hdfs://hadoop10:9000/out234"); // 输出到外部文件


        // 5、执行
        env.execute("我的第一个Flink程序，嘿嘿~");
    }
}
